package top.xphu.component;

import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;
import top.xphu.config.KafkaConfig;
import top.xphu.config.KafkaConfigProperties;

import java.util.concurrent.ExecutionException;

/**
 * @author xphu
 * @version 1.0
 * @date 2021/9/3 15:56
 */
@Component
public class KafkaTemple {

    private final KafkaConfig kafkaConfig;
    private final KafkaProducer<String, String> producer;
    private final KafkaConfigProperties configProperties;

    public KafkaTemple(KafkaConfig kafkaConfig) {
        this.kafkaConfig = kafkaConfig;
        this.producer = kafkaConfig.getProducer();
        this.configProperties = kafkaConfig.getConfigProperties();
    }

    /**
     * 异步发送数据
     * @param message 消息
     */
    public void asyncProduce(String topic, String message){
        try {
            ProducerRecord producerRecord = new ProducerRecord(StringUtils.isEmpty(topic) ? configProperties.getTopic() : topic, message);
            producer.send(producerRecord, (recordMetadata, exception) -> {
                if (recordMetadata != null) {
                    System.out.println(("发送数据信息:partition:" + recordMetadata.partition() + ", offset: " + recordMetadata.offset() + ", timestamp: " + recordMetadata.timestamp()));
                }
            });
        } catch (Exception e) {
            e.printStackTrace();
        }
        producer.flush();
        producer.close();
    }


    /**
     * 异步发送数据
     * @param message 消息
     */
    public void asyncProduce(String topic, String message, Callback callback){
        try {
            ProducerRecord producerRecord = new ProducerRecord(topic, message);
            producer.send(producerRecord,callback);
        } catch (Exception e) {
            e.printStackTrace();
        }
        producer.flush();
        producer.close();
    }

    /**
     * 同步发送
     */
    public RecordMetadata syncProduce(String topic, String message) {
        try {
            ProducerRecord<String, String> producerRecord = new ProducerRecord<>(topic, message);
            RecordMetadata recordMetadata = (RecordMetadata) producer.send(producerRecord).get();
            System.out.println(("发送数据信息:partition:" + recordMetadata.partition() + ", offset: " + recordMetadata.offset() + ", timestamp: " + recordMetadata.timestamp()));
            return recordMetadata;
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
            return null;
        } finally {
            producer.flush();
            producer.close();
        }
    }
}
